package cn.dataling.dag.pojo;

import cn.dataling.dag.enums.DagWorkflowNodeState;
import cn.dataling.dag.task.DagWorkflowNodeTask;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * dag节点
 */
public class DagWorkflowNode {

    /**
     * 节点id
     */
    private String id;
    /**
     * 节点名称
     */
    private String name;
    /**
     * 节点开始时间
     */
    private LocalDateTime startTime;
    /**
     * 节点完成时间
     */
    private LocalDateTime completedTime;
    /**
     * dag工作流节点任务
     */
    private DagWorkflowNodeTask dagWorkflowNodeTask;
    /**
     * 节点元数据
     */
    private Object metaData;
    /**
     * 下一组节点
     */
    private final List<DagWorkflowNode> nextNodes = new LinkedList<>();
    /**
     * 上一组节点
     */
    private final List<DagWorkflowNode> prevNodes = new LinkedList<>();
    /**
     * 工作流节点状体
     */
    private DagWorkflowNodeState state = DagWorkflowNodeState.READY;

    /**
     * 重试次数
     */
    private final AtomicInteger retryTimes = new AtomicInteger(1);

    /**
     * 添加上一组节点
     */
    public DagWorkflowNode prevNodes(DagWorkflowNode... nodes) {
        if (nodes != null && nodes.length > 0) {

            // 取交集 避免重复添加
            List<DagWorkflowNode> waitAddNodes = Arrays.stream(nodes)
                    .filter(e -> prevNodes.stream().noneMatch(n -> n.getId().equals(e.getId())))
                    .collect(Collectors.toList());

            prevNodes.addAll(waitAddNodes);
            for (DagWorkflowNode node : waitAddNodes) {
                node.nextNodes.add(this);
            }
        }
        return this;
    }

    public DagWorkflowNode setId(String id) {
        this.id = id;
        return this;
    }

    public DagWorkflowNode setName(String name) {
        this.name = name;
        return this;
    }

    public DagWorkflowNode setDagWorkflowNodeTask(DagWorkflowNodeTask dagWorkflowNodeTask) {
        this.dagWorkflowNodeTask = dagWorkflowNodeTask;
        return this;
    }

    public DagWorkflowNode setState(DagWorkflowNodeState state) {
        this.state = state;
        return this;
    }

    public DagWorkflowNode setStartTime(LocalDateTime startTime) {
        this.startTime = startTime;
        return this;
    }

    public DagWorkflowNode setMetaData(Object metaData) {
        this.metaData = metaData;
        return this;
    }

    public DagWorkflowNode setCompletedTime(LocalDateTime completedTime) {
        this.completedTime = completedTime;
        return this;
    }

    public DagWorkflowNode setRetryTimes(int retryTimes) {
        this.retryTimes.set(retryTimes);
        return this;
    }

    public String getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    public DagWorkflowNodeTask getDagWorkflowNodeTask() {
        return dagWorkflowNodeTask;
    }

    public List<DagWorkflowNode> getNextNodes() {
        return nextNodes;
    }

    public List<DagWorkflowNode> getPrevNodes() {
        return prevNodes;
    }

    public DagWorkflowNodeState getState() {
        return state;
    }

    public LocalDateTime getStartTime() {
        return startTime;
    }

    public LocalDateTime getCompletedTime() {
        return completedTime;
    }

    public <T> T getMetaData(Class<T> tClass) {
        return  tClass.cast(metaData);
    }

    public AtomicInteger getRetryTimes() {
        return retryTimes;
    }
}
